package kafka

import (
	"fmt"
	"github.com/Shopify/sarama"
)

//新建kafka producer
//
//参数: addr []string
//
//返回 producer sarama.SyncProducer 和 error
func NewProducer(addr []string) (producer sarama.SyncProducer, err error) {
	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForAll
	config.Producer.Partitioner = sarama.NewRandomPartitioner
	config.Producer.Return.Successes = true

	//链接kafka
	producer, err = sarama.NewSyncProducer(addr, config)
	if err != nil {
		fmt.Println("[pkg kafka] producer closed, err:", err)
		return
	}
	fmt.Println("[pkg kafka] connect to kafka success")
	return
}

func NewConsumer(addrs []string) (sarama.Consumer, error) {
	return sarama.NewConsumer(addrs, nil)
}

//将某个消息发送到指定的topic
//
//参数: producer sarama.SyncProducer, topic string, data string
func SendToKafka(producer sarama.SyncProducer, topic, data string) {
	//构造一个消息
	msg := &sarama.ProducerMessage{}
	msg.Topic = topic
	msg.Partition = 2
	msg.Value = sarama.StringEncoder(data)
	//发送消息
	pid, offset, err := producer.SendMessage(msg)
	if err != nil {
		fmt.Println("[pkg kafka] send msg failed, err:", err)
		return
	}
	fmt.Printf("发送成功：pid:%v offset:%v\n", pid, offset)
}

type Data struct {
	Partition int32  `json:"partition"`
	Offset    int64  `json:"offset"`
	Key       string `json:"key"`
	Value     string `json:"value"`
}

/*
从 Kafka 中取消息
*/
func GetFromKafka(consumer sarama.Consumer, topic string, sendToEs func(pc sarama.PartitionConsumer)) error {
	//得到所有的分区列表
	var (
		err               error
		partitionConsumer sarama.PartitionConsumer
		partitions        []int32
	)
	partitions, err = consumer.Partitions(topic)
	if err != nil {
		return err
	}
	fmt.Println("[pkg kafka] partitionList:", partitions)
	//遍历所有分区
	for partition := range partitions {
		//消费分区，从最新的 offset
		fmt.Println(partition)
		partitionConsumer, err = consumer.ConsumePartition(topic, int32(partition), sarama.OffsetNewest)
		if err != nil {
			return err
		}
		// 将 partitionConsumer 传递给回调函数
		go sendToEs(partitionConsumer)
	}
	return nil
}
